package io.vertx.kue.queue

import io.vertx.kue.Kue
import io.vertx.kue.util.RedisHelper

import io.vertx.core.AbstractVerticle
import io.vertx.core.AsyncResult
import io.vertx.core.Future
import io.vertx.core.Handler
import io.vertx.core.eventbus.EventBus
import io.vertx.core.eventbus.MessageConsumer
import io.vertx.core.json.Json
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import io.vertx.core.logging.Logger
import io.vertx.core.logging.LoggerFactory
import io.vertx.redis.RedisClient

import java.util.Optional


/**
 * The verticle for processing Kue tasks.
 *
 * @author Eric Zhao
 */
class KueWorker(private val type: String, private val jobHandler: Handler<Job>, private val kue: Kue) : AbstractVerticle() {
    private var client: RedisClient? = null // Every worker use different clients.
    private var eventBus: EventBus? = null
    private var job: Job? = null

    private var doneConsumer: MessageConsumer<*>? = null // Preserve for unregister the consumer.
    private var doneFailConsumer: MessageConsumer<*>? = null

    /**
     * Get a job from Redis backend by priority.
     *
     * @return async result of job
     */
    private val jobFromBackend: Future<Optional<Job>>
        get() {
            val future = Future.future<Optional<Job>>()
            client!!.blpop(RedisHelper.getKey(this.type + ":jobs"), 0) { r1 ->
                if (r1.failed()) {
                    client!!.lpush(RedisHelper.getKey(this.type + ":jobs"), "1") { r2 ->
                        if (r2.failed())
                            future.fail(r2.cause())
                    }
                } else {
                    this.zpop(RedisHelper.getKey("jobs:" + this.type + ":INACTIVE"))
                            .compose(kue::getJob)
                            .setHandler({ r ->
                                if (r.succeeded()) {
                                    future.complete(r.result())
                                } else
                                    future.fail(r.cause())
                            })
                }
            }
            return future
        }

    @Throws(Exception::class)
    override fun start() {
        this.eventBus = vertx.eventBus()
        this.client = RedisHelper.client(vertx, config())

        prepareAndStart()
    }

    /**
     * Prepare job and start processing procedure.
     */
    private fun prepareAndStart() {
        cleanup()
        this.jobFromBackend.setHandler { jr ->
            if (jr.succeeded()) {
                if (jr.result().isPresent()) {
                    this.job = jr.result().get()
                    process()
                } else {
                    this.emitJobEvent("error", null, JsonObject().put("message", "job_not_exist"))
                    throw IllegalStateException("job not exist")
                }
            } else {
                this.emitJobEvent("error", null, JsonObject().put("message", jr.cause().message))
                jr.cause().printStackTrace()
            }
        }
    }

    /**
     * Process the job.
     */
    private fun process() {
        val curTime = System.currentTimeMillis()
        this.job!!.setStarted_at(curTime)
                .set("started_at", curTime.toString())
                .compose(Job::active)
        .setHandler { r ->
            if (r.succeeded()) {
                val j = r.result()
                // emit start event
                this.emitJobEvent("start", j, null)

                logger.debug("KueWorker::process[instance:Verticle(" + this.deploymentID() + ")] with job " + job!!.getId())
                // process logic invocation
                try {
                    jobHandler.handle(j)
                } catch (ex: Exception) {
                    j.done(ex)
                }

                // subscribe the job done event

                doneConsumer = eventBus!!.consumer<JsonObject>(Kue.workerAddress("done", j)) { msg ->
                    createDoneCallback(j).handle(Future.succeededFuture(
                            msg.body().getJsonObject("result")))
                }
                doneFailConsumer = eventBus!!.consumer<String>(Kue.workerAddress("done_fail", j)) { msg ->
                    createDoneCallback(j).handle(Future.failedFuture(msg.body()))
                }
            } else {
                this.emitJobEvent("error", this.job, JsonObject().put("message", r.cause().message))
                r.cause().printStackTrace()
            }
        }
    }

    private fun cleanup() {
        doneConsumer?.let{ it.unregister() }
        doneFailConsumer?.let{ it.unregister() }
//        Optional.ofNullable(doneConsumer).ifPresent({ it.unregister() })
//        Optional.ofNullable(doneFailConsumer).ifPresent({ it.unregister() })
        this.job = null
    }

    private fun error(ex: Throwable, job: Job) {
        val err = JsonObject().put("message", ex.message)
                .put("id", job.getId())
        eventBus!!.send(Kue.workerAddress("error"), err)
    }

    private fun fail(ex: Throwable) {
        job!!.failedAttempt(ex).setHandler({ r ->
            if (r.failed()) {
                this.error(r.cause(), job!!)
            } else {
                val res = r.result()
                if (res.hasAttempts()) {
                    this.emitJobEvent("failed_attempt", job, JsonObject().put("message", ex.message)) // shouldn't include err?
                } else {
                    this.emitJobEvent("failed", job, JsonObject().put("message", ex.message))
                }
                prepareAndStart()
            }
        })
    }

    /**
     * Redis zpop atomic primitive with transaction.
     *
     * @param key redis key
     * @return the async result of zpop
     */
    private fun zpop(key: String): Future<Long> {
        val future = Future.future<Long>()
        client!!.transaction()
                .multi(_failure())
                .zrange(key, 0, 0, _failure())
                .zremrangebyrank(key, 0, 0, _failure())
                .exec { r ->
                    if (r.succeeded()) {
                        val res = r.result()
                        if (res.getJsonArray(0).size() == 0)
                        // empty set
                            future.fail(IllegalStateException("Empty zpop set"))
                        else {
                            try {
                                future.complete(java.lang.Long.parseLong(RedisHelper.stripFIFO(
                                        res.getJsonArray(0).getString(0))))
                            } catch (ex: Exception) {
                                future.fail(ex)
                            }

                        }
                    } else {
                        future.fail(r.cause())
                    }
                }
        return future
    }

    private fun createDoneCallback(job: Job?): Handler<AsyncResult<JsonObject>> {
        return Handler{ r0 ->
            if (job == null) {
                // maybe should warn
//                return
            }
            if (r0.failed()) {
                this.fail(r0.cause())
//                return
            }
            val dur = System.currentTimeMillis() - job!!.getStarted_at()
            job!!.setDuration(dur)
                    .set("duration", dur.toString())
            val result = r0.result()
            if (result != null) {
                job!!.setResult(result)
                        .set("result", result!!.encodePrettily())
            }

            job!!.complete().setHandler({ r ->
                if (r.succeeded()) {
                    val j = r.result()
                    if (j.isRemoveOnComplete()) {
                        j.remove()
                    }
                    this.emitJobEvent("complete", j, null)

                    this.prepareAndStart() // prepare for next job
                }
            })
        }
    }

    @Throws(Exception::class)
    override fun stop() {
        // stop hook
        cleanup()
    }

    /**
     * Emit job event.
     *
     * @param event event type
     * @param job   corresponding job
     * @param extra extra data
     */
    private fun emitJobEvent(event: String, job: Job?, extra: JsonObject?) {
        val data = JsonObject().put("extra", extra)
        if (job != null) {
            data.put("job", job!!.toJson())
        }
        eventBus!!.send(Kue.workerAddress("job_" + event), data)
        when (event) {
            "failed", "failed_attempt" -> eventBus!!.send(Kue.getCertainJobAddress(event, job!!), data)
            "error" -> eventBus!!.send(Kue.workerAddress("error"), data)
            else -> eventBus!!.send(Kue.getCertainJobAddress(event, job!!), job!!.toJson())
        }
    }

    companion object {

        private val logger = LoggerFactory.getLogger(Job::class.java!!)

        private fun <T> _failure(): Handler<AsyncResult<T>> {
            return Handler{ r ->
                if (r.failed())
                    r.cause().printStackTrace()
            }
        }
    }

}
